package com.xctech.yace.thread;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonObject;
import com.xctech.yace.util.TimeUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;

public class ConsumerTpsThread implements Runnable{

    private KafkaConsumer<String,String> consumer;

    private ConsumerRecords<String, String> msgs;

    private final String[] topics;

    public ConsumerTpsThread(KafkaConsumer consumer, String[] topicName) {
        this.consumer = consumer;
        this.topics = topicName;
        this.consumer.subscribe(Arrays.asList(topicName));
    }

    public static Long minTime = 0l;
    public static Long maxTime = 0l;

    @Override
    public void run() {
        System.out.println("-----------tps计算-------------");
        try{
            for(;;){
                msgs = consumer.poll(1);
                if(null != msgs && msgs.count() > 0){
                    for (ConsumerRecord<String,String> record : msgs){
                        String key = record.key();
                        String[] times = key.split("\\|");
                        JSONObject value = (JSONObject) JSONObject.parse(record.value());
                        Long tmpTime = Long.valueOf(times[2]);
                        Double interval = 0.0;
                        String number = times[0];

                        if("1".equals(number)){
                            minTime = value.getLong("OCCU_TIME");
                        }else{
                            maxTime = tmpTime > maxTime ? tmpTime : maxTime;
                            interval = TimeUtil.calIntervalMicro2Second(minTime, maxTime);
                        }

                        //计算公式 =  xid / cals(本笔的OCCU_TIME微秒 - xid为0的OCCU_TIME微秒)
                        Double tps = 0.0;
                        if(interval != 0.0 && number!= null){
                           tps = Integer.parseInt(number) * 1.0 / interval;
                        }
                        System.out.println("=======receive:offset=="+record.offset()+" tps = "+tps+"--- 数量 = " + number +"--- 开始时间 = "+minTime +"--- 结束时间 = "+maxTime);
                    }
                }else {
                    Thread.sleep(1);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }





}
